﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Runtime.Caching;
using System.Text;
using System.Threading.Tasks;
using Fasterflect;
using Kugar.Core.BaseStruct;
using Kugar.Core.ExtMethod;
using Kugar.Tool.RemoteEventBus.Enums;
using Kugar.Tool.RemoteEventBus.Triggers;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Framing;
using Kugar.Tool.RemoteEventBus.Helpers;

namespace Kugar.Tool.RemoteEventBus
{
    public class SyncEventTrigger<T> : ISyncEventTrigger<T>, IEventTriggerSubscriberHandler // where T : EventArgs
    {
        private RemoteEventBus _bus = null;
        private string _eventName = "";
        private IModel _channel = null;
        private EventingBasicConsumer _consumer = null;
        private string _queueName = "";
        private HashSet<string> _subsecribeQueue = new HashSet<string>();
        private string _eventRelyRouteKey = "";
        private ConcurrentDictionary<string, SyncEventRelyAwatiable> _awaiters = new ConcurrentDictionary<string, SyncEventRelyAwatiable>();


        internal SyncEventTrigger(RemoteEventBus bus, string eventName)
        {
            _bus = bus;
            _eventName = eventName;
            _channel = bus.Connection.CreateModel();

            _eventRelyRouteKey = $"Sync_{eventName}_Rely_{bus.ServiceID}";

            _queueName = $"Sync_{eventName}_{bus.ServiceID}";

            _consumer = new EventingBasicConsumer(_channel);

            _consumer.Received += consumer_Received;
            _consumer.Shutdown += consumer_Shutdown;

            _channel.BasicReturn += channel_BasicReturn;
            _channel.QueueDeclare(_queueName, false, false, true, null);
            _channel.QueueBind(_queueName, "RemoteEventBus_Sync", _eventRelyRouteKey, null);
            _channel.BasicConsume(_queueName, false, _consumer);
        }

        /// <summary>
        /// 当前事件被触发时,触发该事件
        /// </summary>
        public event EventHandler<EventTriggerEventArgs> TriggerEvent;

        /// <summary>
        /// 当订阅者已处理该事件的时候,触发
        /// </summary>
        public event EventHandler<ClientEventHandeEventArgs> ClientHandledEvent;

        public EventTypeEnum Type => EventTypeEnum.Async;

        public IRemoteEventBus EventBus => _bus;

        public string EventName => _eventName;

        public IEnumerable<string> SubscriberClientIDs => _subsecribeQueue;

        void IEventTriggerSubscriberHandler.Close()
        {
            _channel.Close();
        }

        int IEventTriggerSubscriberHandler.AddSubscriber(string queueName)
        {
            _subsecribeQueue.Add(queueName);

            return _subsecribeQueue.Count;
        }

        int IEventTriggerSubscriberHandler.RemoveSubscriber(string queueName)
        {
            _subsecribeQueue.Remove(queueName);

            return _subsecribeQueue.Count;
        }

        void IEventTriggerSubscriberHandler.ClearSubscriber()
        {
            _subsecribeQueue.Clear();
        }

        public JObject[] Trigger(T eventArgs)
        {
            return Trigger(eventArgs, 15000, true);
        }


        /// <summary>
        /// 同步方式触发同步事件
        /// </summary>
        /// <param name="eventArgs"></param>
        /// <param name="timeout">超时时间(毫秒)</param>
        /// <param name="isThrowErrorWhenTimeout">当远程调用出现timeout的时候,是否要抛出timeout错误,如果为false,则返回已接收到的数值,不抛出错误,,,默认为true</param>
        /// <returns></returns>
        public JObject[] Trigger(T eventArgs, int timeout, bool isThrowErrorWhenTimeout)
        {
            return TriggerAsync(eventArgs, timeout, isThrowErrorWhenTimeout).Result;
        }

        /// <summary>
        /// 异步触发同步事件
        /// </summary>
        /// <param name="eventArgs"></param>
        /// <returns></returns>
        public Task<JObject[]> TriggerAsync(T eventArgs)
        {
            return TriggerAsync(eventArgs, 15000, true);
        }

        /// <summary>
        /// 异步触发同步事件
        /// </summary>
        /// <param name="eventArgs"></param>
        /// <param name="timeout">超时时间(毫秒)</param>
        /// <param name="isThrowErrorWhenTimeout">当远程调用出现timeout的时候,是否要抛出timeout错误,如果为false,则返回已接收到的数值,不抛出错误,,,默认为true</param>
        /// <returns></returns>
        public async Task<JObject[]> TriggerAsync(T eventArgs, int timeout, bool isThrowErrorWhenTimeout)
        {
            return await publishToRouteKey($"{_eventName}_Sync_Trigger", eventArgs,timeout,isThrowErrorWhenTimeout);

        }

        public JObject[] TriggerToClientID(string clientID, T eventArgs)
        {
            return TriggerToClientIDAsync(clientID, eventArgs).Result;
        }

        public async Task<JObject[]> TriggerToClientIDAsync(string clientID, T eventArgs)
        {
            return await publishToRouteKey($"Sync_{_eventName}_Client_{clientID}", eventArgs);
        }


        private void consumer_Shutdown(object sender, ShutdownEventArgs e)
        {
            _consumer.Received -= consumer_Received;
            _consumer.Shutdown -= consumer_Shutdown;
        }

        private void consumer_Received(object sender, BasicDeliverEventArgs e)
        {


            if (e.RoutingKey != _eventRelyRouteKey ||
                e.BasicProperties?.Type != "SyncEventRely"
                )
            {
                _channel.BasicAck(e.DeliveryTag, false);
                return;
            }

            SyncEventRelyPackage package;
            SyncEventRelyAwatiable awaiter;

            try
            {
                var json = JObject.Parse(Encoding.UTF8.GetString(e.Body));

                package = json.ToObject<SyncEventRelyPackage>();

                if (!_awaiters.TryGetValue(package.TriggerID, out awaiter))
                {
                    return;
                }
            }
            catch (Exception)
            {

                throw;
            }
            finally
            {
                _channel.BasicAck(e.DeliveryTag, false);
            }

            if (awaiter.SetResult(
                    package.ClientID,
                    package.ReturnValue.HasData() ? package.ReturnValue : null)
                )
            {
                _awaiters.TryRemove(package.TriggerID);
            }

            //TODO:触发完成
            if (ClientHandledEvent != null)
            {
                ClientHandledEvent(this, new ClientEventHandeEventArgs(package.ClientID, _eventName, package.TriggerID, EventRelyType.Normal, null, package.ReturnValue));
            }
        }

        private void channel_BasicReturn(object sender, BasicReturnEventArgs e)
        {
            SyncEventRelyAwatiable awaiter;

            if (_awaiters.TryRemove(e.BasicProperties.MessageId, out awaiter))
            {
                awaiter.SetResult(new JObject[0], true);

                if (ClientHandledEvent != null)
                {
                    ClientHandledEvent(this, new ClientEventHandeEventArgs("", _eventName, e.BasicProperties.MessageId, EventRelyType.NoSubscriber));
                }
            }
        }

        private async Task<JObject[]> publishToRouteKey<T>(string routeKey, T eventArgs,int timeout=15000, bool isThrowErrorWhenTimeout = true)
        {
            if (!_subsecribeQueue.HasData()) //如果没有任何订阅者,就直接返回空数组
            {
                return new JObject[0];
            }

            var triggerID = Guid.NewGuid().ToString();

            var item = new SyncEventRelyAwatiable(timeout, new HashSet<string>(_subsecribeQueue), isThrowErrorWhenTimeout);

            _awaiters.TryAdd(triggerID, item);

            var args = JsonConvert.SerializeObject(new SyncEventRaisePackage()
            {
                Data = JObject.FromObject(eventArgs), // JsonConvert.SerializeObject(eventArgs),
                EventName = _eventName,
                TypeName = eventArgs.GetType().FullName,
                RelyQueueName = _eventRelyRouteKey
            });

            _channel.BasicPublish("RemoteEventBus_Sync",
                routeKey,
                true,
                new BasicProperties()
                {
                    Expiration = "15000",
                    Type = _eventName,
                    MessageId = triggerID,
                    ReplyTo = _eventRelyRouteKey,
                    CorrelationId = _bus.ServiceID
                },
                Encoding.UTF8.GetBytes(args)
                    );

            if (TriggerEvent != null)
            {
                TriggerEvent(this, new EventTriggerEventArgs(triggerID, _eventName, args, routeKey));
            }

            try
            {
                var ret = await item;

                return ret.ToArrayEx();
            }
            catch (TimeoutException)
            {
                if (!isThrowErrorWhenTimeout)
                {
                    return item.GetResult();
                }
                else
                {
                    throw;
                }
            }
            catch (Exception)
            {
                throw;
            }
            finally
            {
                _awaiters.TryRemove(triggerID);
            }
        }



    }
}